package org.example.snow.demo3;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

/**
 * @author snowsong
 */
public class FlinkServer extends Thread{
    @Override
    public void run() {
            String ip = "0.0.0.0";
            int port = 8886;
            StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
            // 使用socket流创建一个从 socket 读取文本的数据流，以换行符 \n 作为分隔符
            DataStreamSource<String> stringDataStreamSource = executionEnvironment.socketTextStream(ip, port, "\n");
            SingleOutputStreamOperator<Tuple2<String, Long>> tuple2SingleOutputStreamOperator = stringDataStreamSource.flatMap(new FlatMapFunction<String, Tuple2<String, Long>>() {

                @Override
                public void flatMap(String s, Collector<Tuple2<String, Long>> collector) throws Exception {
                    String[] splits = s.split("\\s");
                    for (String word : splits) {
                        collector.collect(Tuple2.of(word, 1L));
                    }
                }
            });
            SingleOutputStreamOperator<Tuple2<String, Long>> word = tuple2SingleOutputStreamOperator
                    .keyBy(new KeySelector<Tuple2<String, Long>, Object>() {
                        @Override
                        public Object getKey(Tuple2<String, Long> stringLongTuple2) throws Exception {
                            return stringLongTuple2.f0;
                        }
                    })
                    .window(SlidingProcessingTimeWindows.of(Time.seconds(5), Time.seconds(1)))
                    .sum(1);
            word.print("======");
            try {
                executionEnvironment.execute("stream!");
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }

}
